package com.zzw.demo.util;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;

public class MySinkHbase extends RichSinkFunction<String> {
    private Connection conn;
    //创建连接
    @Override
    public void open(Configuration parameters) throws Exception {
        org.apache.hadoop.conf.Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "hadoop-single:2181");
        conn = ConnectionFactory.createConnection(config);
    }
    //关流
    @Override
    public void close() throws Exception {
        conn.close();
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        // 获取表  表格
        Table table = conn.getTable(TableName.valueOf("exam"));
        // 创建put对象  行建
        Put put = new Put("001".getBytes());
        // 添加字段,存入数据  列祖
        put.addColumn("info".getBytes(), "data".getBytes(), value.getBytes());
        // 提交修改
        table.put(put);
    }
}
